package com.kafka_test.app;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;

import com.alibaba.fastjson.JSONObject;
import com.kafka_test.app.service.OffsetService;
import org.mybatis.spring.annotation.MapperScan;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
@MapperScan("com.kafka_test.app.mapper")
public class MysqlToKafka {

    private static OffsetService offsetService;

    @Autowired
    public MysqlToKafka(OffsetService offsetService) {
        MysqlToKafka.offsetService = offsetService;
    }

    public static long queryTableData(String tableName, long offset, int batchSize, List<String> resultList) {
        long maxOffset = offset;
        List<Map<String, Object>> re = offsetService.queryTableData(tableName, offset);
        for (int i = 0; i < batchSize && i < re.size(); i++) {
            //resultList.add(re.get(i).toString());
            resultList.add(String.valueOf(new JSONObject(re.get(i))));
            offset = Long.parseLong(String.valueOf(re.get(i).get("id_producer")));
            maxOffset = Math.max(maxOffset, offset);
        }
        return maxOffset;
    }

    public static void run(String tableName, int batchSize, String topic) {
        //TODO 获取offset
        Object obj = offsetService.getOffset(tableName);
        long offset = 0;
        //TODO 初始化offset
        if (obj == null) {
            offsetService.initializeOffset(tableName, offset);
            System.out.println("初始化表" + tableName + "的offset:" + offset);
        }
        else{
            offset = Long.parseLong(obj.toString());
        }
        //TODO 实时采集没有获取的数据
        List<String> resultList = new ArrayList<>();
        offset = queryTableData(tableName, offset, batchSize, resultList);
        System.out.println("本批次查询表" + tableName + "的最大offset:" + offset);
        //TODO 发送kafka
        offsetService.sendMessage(resultList, topic);
        System.out.println("数据发送kakfa成功");
        //TODO 更新offset
        offsetService.updateOffset(tableName, offset);
        System.out.println("更新表" + tableName + "的offset:" + offset);
    }

    public static void main(String[] args) {
        SpringApplication.run(MysqlToKafka.class, args);
        while (true) {
            System.out.println("run() start...");
            run("producer", 100, "producer_topic");
            System.out.println("run() stop..\n");
            try {
                Thread.sleep(6000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

}
